package com.hna.cums.web.MsgTsskCtl;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hna.cums.web.loginctl.LoginController;

@Controller
@RequestMapping("/MsgTaskCtl/*")
public class MsgCtl {
	/** 日志 */
    static final Logger logger = LoggerFactory.getLogger(MsgCtl.class);
    static final String SendTopic = "test";
    //static final Properties props;
    @ResponseBody
    @RequestMapping("TestInputMsg.do")
    public String TestInputMsg(HttpServletRequest request, HttpServletResponse response){
    
    	logger.info("TestInputMsg.do begin!");
    	try {
			request.setCharacterEncoding("utf-8");
			String ImageID = request.getParameter("imageid");
			String ClickUser = request.getParameter("clickuser");
			String MsgID = "00000001";
			/*初始化消息队列*/
			Properties props = new Properties();
			props.put("bootstrap.servers", "10.71.200.109:9092");
			logger.debug("zookeeper2.connect: 10.71.200.109:2182");
			props.put("zookeeper.connect", "10.71.200.109:2182");
//			props.put("broker.list", "10.71.200.109:9092");
			props.put("client.id", "DataPlt-test-client");
			props.put("acks", "1");
			props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
			props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
			
			/*组报文*/
			//JSONObject jsonObject = new JSONObject();
		
			HashMap<String, String> Msg = new HashMap<String, String>();
			Msg.put("MsgID", MsgID);
			Msg.put("ImageID", ImageID);
			Msg.put("ClickUser", ClickUser);
		    String jsonstr = JSON.toJSONString(Msg);
			
			
			/*放入测试消息*/
			Producer<String, String> producer = new KafkaProducer<String, String>(props);
			producer.send(new ProducerRecord<String, String>(SendTopic, MsgID, jsonstr));
			
			
			response.getWriter().print( "200");
			response.getWriter().flush();
			response.getWriter().close();
			logger.debug("MsgCt: "+ jsonstr);
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    	
    	return null;	
	}
}
